package com.ds.infrastructure.hyperspace.container.endpoint.connector;

import com.ds.infrastructure.hyperspace.common.EndpointNode;
import com.ds.infrastructure.hyperspace.common.TargetEndpoint;
import com.ds.infrastructure.hyperspace.common.TargetEndpointStream;
import com.ds.infrastructure.hyperspace.common.config.TargetEndpointConfig;
import com.ds.infrastructure.hyperspace.common.config.TargetNodeConfig;
import com.ds.infrastructure.hyperspace.container.endpoint.EndpointInitException;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.NettyTargetEndpoint;
import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.flow.HyperspaceDuplexFlowPipeline;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceTargetNodeContext;
import com.ds.infrastructure.hyperspace.container.flow.operation.EntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.flow.operation.TargetFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import io.netty.channel.EventLoop;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.ds.infrastructure.hyperspace.common.StateEnum.*;
import static com.ds.infrastructure.hyperspace.common.TargetEndpointStreamStatus.INIT;
import static com.ds.infrastructure.hyperspace.common.TargetEndpointStreamType.PERMANENT;
import static java.util.concurrent.TimeUnit.MINUTES;

/**
 * @author  Weishaoying
 */
@Slf4j
public class HyperspaceConnector {

    private static final int STREAM_FORCE_SHUTDOWN_DELAY = 2;

    private static final long CONTEXT_PENDING_DELAY = 100L;
    private static final long MAX_CONTEXT_PENDING_DELAY = 2000L;

    private final NettyContainer nettyContainer;

    @Getter
    private final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

    private final TargetFlowOperation<HyperspaceDuplexFlowContext> targetFlowOperation;

    private final EventLoop entryEventLoop;

    private final Map<String, DefaultTargetNodeWrapper> nodeWrapperMap;

    private final LinkedList<HyperspaceDuplexFlowContext> pendingDataLinkedList;

    private final HyperspaceStreamTimer hyperspaceStreamTimer;

    private final PendingTask pendingTask;

    public HyperspaceConnector(NettyContainer nettyContainer, EventLoop entryEventLoop,
                               EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation,
                               TargetFlowOperation<HyperspaceDuplexFlowContext> targetFlowOperation) {
        this.nettyContainer = nettyContainer;
        this.entryFlowOperation = entryFlowOperation;
        this.targetFlowOperation = targetFlowOperation;
        this.entryEventLoop = entryEventLoop;
        this.nodeWrapperMap = new HashMap<>(16);
        this.pendingDataLinkedList = new LinkedList<>();
        long requestCheckPeriod = nettyContainer.getResourceConfig().getTimeoutCheckPeriod();
        this.hyperspaceStreamTimer = new HyperspaceStreamTimer(entryEventLoop,
                requestCheckPeriod, requestCheckPeriod);
        this.pendingTask = new PendingTask();
    }

    public EventLoop eventLoop() {
        return entryEventLoop;
    }

    public void initTargetNodeStreamManager(TargetEndpointNode targetEndpointNode) {
        nodeWrapperMap.putIfAbsent(targetEndpointNode.getKey(), new DefaultTargetNodeWrapper(
                new HyperspaceTargetNodeContext(targetEndpointNode),
                new NettyEndpointNodeStreamManager(nettyContainer, entryEventLoop,
                        targetEndpointNode, this, hyperspaceStreamTimer)
        ));
    }

    public void removeTargetNodeStreamManager(TargetEndpointNode targetEndpointNode) {
        DefaultTargetNodeWrapper targetNodeWrapper = nodeWrapperMap.get(targetEndpointNode.getKey());
        if (targetNodeWrapper != null) {
            entryEventLoop.schedule(targetNodeWrapper.getNodeStreamManager()::shutdownNodeManager,
                    STREAM_FORCE_SHUTDOWN_DELAY, MINUTES);
            nodeWrapperMap.remove(targetEndpointNode.getKey());
        }
    }

    public void initTargetNodeStream(TargetEndpointNode targetEndpointNode, int poolSize) {
        DefaultTargetNodeWrapper targetNodeWrapper = nodeWrapperMap.get(targetEndpointNode.getKey());
        if (targetNodeWrapper == null) {
            throw new EndpointInitException("Stream manager not found!");
        }
        targetNodeWrapper.getNodeStreamManager().initNettyTargetEndpointStreamPool(poolSize);
    }

    public void recoverNodeStreamPool(TargetEndpointNode targetEndpointNode) {
        DefaultTargetNodeWrapper targetNodeWrapper = nodeWrapperMap.get(targetEndpointNode.getKey());
        if (targetNodeWrapper == null) {
            return;
        }
        targetNodeWrapper.getNodeStreamManager().recoverClosedStreamPool();
    }

    public void handleInput(HyperspaceDuplexFlowPipeline pipeline, NettyTargetEndpoint targetEndpoint,
                            HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        if (duplexFlowContext.isClosed() || duplexFlowContext.isPending()) {
            return;
        }

        TargetEndpointStream targetEndpointStream = duplexFlowContext.getCurrentReferTargetEndpointStream();
        if (targetEndpointStream != null) {
            targetEndpointStream.handleInput(duplexFlowContext, data);
            return;
        }

        if (data != duplexFlowContext.getRequestStartState()) {
            entryFlowOperation.interruptAndCloseContext(duplexFlowContext, TARGET_ENDPOINT_STREAM_UNAVAILABLE,
                    "Target stream not found for zone " + duplexFlowContext.getZoneId());
            return;
        }

        int maxPendingSize = targetEndpoint.getEndpointConfig().getMaxPendingRequestSize();
        if (nettyContainer.getTargetEndpointPendingSize(targetEndpoint.getNamespace(),
                duplexFlowContext.getZoneId()) >= maxPendingSize) {
            entryFlowOperation.interruptAndCloseContext(duplexFlowContext, REQUEST_NOT_ACCEPTABLE,
                    "Target endpoint (" + targetEndpoint.getNamespace() +
                            ") exceed max pending request size " + maxPendingSize);
            return;
        }

        List<EndpointNode<TargetNodeConfig>> nodeList = targetEndpoint.getEndpointNodes(duplexFlowContext.getZoneId());
        if (CollectionUtils.isEmpty(nodeList)) {
            entryFlowOperation.interruptAndCloseContext(duplexFlowContext, TARGET_ENDPOINT_NODE_NOT_FOUND,
                    "Target node list not found for zone " + duplexFlowContext.getZoneId());
            return;
        }
        EndpointNode<TargetNodeConfig> targetEndpointNode = pipeline.selectNode(duplexFlowContext, nodeWrapperMap);
        if (targetEndpointNode == null || targetEndpointNode.isEmptyNodes()) {
            entryFlowOperation.interruptAndCloseContext(duplexFlowContext, TARGET_ENDPOINT_NODE_NOT_FOUND,
                    "Load balancer can't selected a suitable node for zone " +
                            duplexFlowContext.getZoneId());
            return;
        }
        DefaultTargetNodeWrapper targetNodeWrapper = nodeWrapperMap.get(targetEndpointNode.getKey());
        if (targetNodeWrapper == null) {
            entryFlowOperation.interruptAndCloseContext(duplexFlowContext, TARGET_ENDPOINT_STREAM_UNAVAILABLE,
                    "Target endpoint node manager was not found!");
            return;
        }

        NettyEndpointNodeStreamManager nodeStreamManager = targetNodeWrapper.getNodeStreamManager();
        if (duplexFlowContext.getEndpointProtocol().isLongLive()) {
            TargetEndpointStream newTargetEndpointStream = nodeStreamManager.createTargetEndpointStream(
                    PERMANENT, true);
            if (newTargetEndpointStream.getStatus().getCodeNum() < INIT.getCodeNum()) {
                String message = "Create target endpoint stream failed!";
                if (newTargetEndpointStream instanceof EmptyTargetEndpointStream) {
                    message += "target connections exceed max node size or max global size";
                }
                entryFlowOperation.interruptAndCloseContext(duplexFlowContext, TARGET_ENDPOINT_UNAVAILABLE, message);
            } else {
                duplexFlowContext.setReferTargetEndpointStream(newTargetEndpointStream);
                newTargetEndpointStream.handleInput(duplexFlowContext, data);
            }
        } else {
            TargetEndpointStream acquiredTargetEndpointStream = nodeStreamManager
                    .acquireTargetEndpointStream();
            if (acquiredTargetEndpointStream.getStatus().getCodeNum() < INIT.getCodeNum()) {
                addShortLivePendingContext(duplexFlowContext);
            } else {
                duplexFlowContext.setReferTargetEndpointStream(acquiredTargetEndpointStream);
                acquiredTargetEndpointStream.handleInput(duplexFlowContext, data);
            }
        }
    }

    public void handleInputComplete(HyperspaceDuplexFlowContext duplexFlowContext) {
        TargetEndpointStream targetEndpointNodeStream = duplexFlowContext.getCurrentReferTargetEndpointStream();
        if (duplexFlowContext.isClosed()) {
            return;
        }
        if (targetEndpointNodeStream != null) {
            targetEndpointNodeStream.handleInputComplete(duplexFlowContext);
        } else {
            addShortLivePendingContext(duplexFlowContext);
        }
    }

    public void onStreamRelease() {
        pendingTask.run();
    }

    public void onContextWriteFailed(HyperspaceDuplexFlowContext shortLiveDuplexFlowContext) {
        shortLiveDuplexFlowContext.setReferTargetEndpointStream(null);
        addShortLivePendingContext(shortLiveDuplexFlowContext);
    }

    private void addShortLivePendingContext(HyperspaceDuplexFlowContext duplexFlowContext) {
        if (duplexFlowContext.getEndpointProtocol().isLongLive() ||
                duplexFlowContext.isClosed() || duplexFlowContext.isPending()) {
            return;
        }
        if (duplexFlowContext.checkTimeout()) {
            duplexFlowContext.onTimeout();
            return;
        }

        TargetEndpoint<TargetEndpointConfig> targetEndpoint = duplexFlowContext.getTargetEndpoint();
        int maxPendingSize = targetEndpoint.getEndpointConfig().getMaxPendingRequestSize();
        if (nettyContainer.getTargetEndpointPendingSize(targetEndpoint.getNamespace(),
                duplexFlowContext.getZoneId()) >= maxPendingSize) {
            entryFlowOperation.interruptAndCloseContext(duplexFlowContext, REQUEST_NOT_ACCEPTABLE,
                    "Target endpoint (" + targetEndpoint.getNamespace() +
                            ") exceed max pending request size " + maxPendingSize);
            return;
        }
        duplexFlowContext.touchForPending();
        duplexFlowContext.setPending(true);
        pendingDataLinkedList.offer(duplexFlowContext);
        targetFlowOperation.handlePendingRequestIncrement(
                duplexFlowContext.getTargetEndpoint().getNamespace(), duplexFlowContext.getZoneId());
        entryEventLoop.schedule(pendingTask, CONTEXT_PENDING_DELAY, TimeUnit.MILLISECONDS);
    }

    class PendingTask implements Runnable {

        private long noTargetNodePendingDelayTime = 50L;

        private long getNoTargetNodePendingDelayTime() {
            noTargetNodePendingDelayTime += noTargetNodePendingDelayTime;
            return Math.min(noTargetNodePendingDelayTime, MAX_CONTEXT_PENDING_DELAY);
        }

        @Override
        public void run() {
            if (pendingDataLinkedList.isEmpty()) {
                return;
            }
            HyperspaceDuplexFlowContext pendingDuplexFlowContext = pendingDataLinkedList.peek();
            if (pendingDuplexFlowContext == null) {
                return;
            }
            pendingDuplexFlowContext.endPendingDuration();
            if (pendingDuplexFlowContext.isClosed()) {
                pollPendingData(pendingDuplexFlowContext);
                return;
            }
            if (pendingDuplexFlowContext.checkTimeout()) {
                pollPendingData(pendingDuplexFlowContext);
                pendingDuplexFlowContext.onTimeout();
                return;
            }
            if (!pendingDuplexFlowContext.getEntryProtocolAdapter().isChannelOpen()) {
                pollPendingData(pendingDuplexFlowContext);
                entryFlowOperation.interruptAndCloseContext(pendingDuplexFlowContext,
                        CLIENT_CLOSED_CONNECTION,
                        "Client has closed connection! ");
                return;
            }
            if (!pendingDuplexFlowContext.isInputComplete()) {
                entryEventLoop.schedule(pendingTask, CONTEXT_PENDING_DELAY, TimeUnit.MILLISECONDS);
                return;
            }

            EndpointNode<TargetNodeConfig> targetEndpointNode = pendingDuplexFlowContext.getFlowHandlerPipeline()
                    .selectNode(pendingDuplexFlowContext, nodeWrapperMap);
            DefaultTargetNodeWrapper targetNodeWrapper = nodeWrapperMap.get(targetEndpointNode.getKey());
            if (targetNodeWrapper == null) {
                pollPendingData(pendingDuplexFlowContext);
                entryFlowOperation.interruptAndCloseContext(pendingDuplexFlowContext,
                        TARGET_ENDPOINT_STREAM_UNAVAILABLE,
                        "Target endpoint node manager was not found!");
                return;
            }
            TargetEndpointStream acquiredTargetEndpointStream = targetNodeWrapper.getNodeStreamManager()
                    .acquireTargetEndpointStream();
            if (acquiredTargetEndpointStream.getStatus().getCodeNum() < INIT.getCodeNum()) {
                entryEventLoop.schedule(pendingTask, pendingTask.getNoTargetNodePendingDelayTime(), TimeUnit.MILLISECONDS);
            } else {
                pollPendingData(pendingDuplexFlowContext);
                pendingDuplexFlowContext.setReferTargetEndpointStream(acquiredTargetEndpointStream);
                pendingDuplexFlowContext.setPending(false);
                for (Object data : pendingDuplexFlowContext.getReferDataLinkList()) {
                    acquiredTargetEndpointStream.handleInput(pendingDuplexFlowContext, data);
                }
                acquiredTargetEndpointStream.handleInputComplete(pendingDuplexFlowContext);
            }
        }

        private void pollPendingData(HyperspaceDuplexFlowContext pendingDuplexFlowContext) {
            pendingDataLinkedList.poll();
            targetFlowOperation.handlePendingRequestDecrement(
                    pendingDuplexFlowContext.getTargetEndpoint().getNamespace(),
                    pendingDuplexFlowContext.getZoneId());
        }
    }

}
